package com.geek.framework.study.event;

import com.geek.framework.study.event.zookeeper.ZookeeperDistributedEventPublisher;
import com.geek.framework.study.event.zookeeper.ZookeeperMessagingEngineContext;

import java.util.Date;
import java.util.EventObject;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @Description {@link ZookeeperDistributedEventPublisher}测试类（测试前请先确保本地的zookeeper是否启动监听到了2181端口）
 */
public class ZookeeperDistributedEventPublisherTester {

    public static void main(String[] args) throws InterruptedException {

        ZookeeperMessagingEngineContext context = new ZookeeperMessagingEngineContext("/queue-test");

        // 模拟分布式应用事件消费者
        startConsumer(context);

        // 消息事件发布者
        startProducer(context);
    }

    private static void startConsumer(ZookeeperMessagingEngineContext context) {
        int terms = 3;
        // 循环n次，模拟不同应用的消费
        IntStream.range(0, terms).forEach(seq -> {
            Executors.newSingleThreadExecutor().submit(() -> {
                ZookeeperDistributedEventPublisher distributedEventPublisher = new ZookeeperDistributedEventPublisher(context);
                String applicationName = "app-" + seq;
                distributedEventPublisher.addEventListener(new EventListener() {
                    @Override
                    public void onEvent(final EventObject event) {
                        System.out.printf("Thread[%d] %s receive an event at %s, the event  is %s[Source : %s]\n",
                                Thread.currentThread().getId(),
                                applicationName,
                                new Date(),
                                event.getClass().getName(),
                                event.getSource());
                    }
                });
            });
        });
    }

    private static void startProducer(ZookeeperMessagingEngineContext context) {
        ZookeeperDistributedEventPublisher distributedEventPublisher = new ZookeeperDistributedEventPublisher(context);

        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(() -> {
            distributedEventPublisher.publish(new MessageInfo());
            try {
                TimeUnit.SECONDS.sleep(1L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println();
        }, 1, 2, TimeUnit.SECONDS);

    }
}
